package cn.ymatrix.utils;

import cn.ymatrix.apiclient.DataPostListener;
import cn.ymatrix.apiclient.MxClient;
import cn.ymatrix.apiclient.Result;
import cn.ymatrix.builder.ConnectionListener;
import cn.ymatrix.builder.MxBuilder;
import cn.ymatrix.builder.RequestType;
import cn.ymatrix.data.Column;
import cn.ymatrix.data.Tuple;

import java.util.Map;

public class MxgateBuilder {

    public static void main(String[] args) {
        MxgateBuilder mb = new MxgateBuilder();
        //System.out.println("dddd");
        mb.sd();

    }

    private void sd() {
        MxBuilder builder = MxBuilder.newBuilder()
                .withDropAll(false) // 如果需用于测试，则可设置为 true，不发送数据给 mxgate，直接 drop；发送数据到 mxgate 需设置为 false
                .withCacheCapacity(100000) // 用于暂存 tuples 微批的 queue的大小
                .withCacheEnqueueTimeout(2000) // 若queue满，tuples 存入 queue 的等待超时时间。若超时，会抛出IllegalStateException
                .withConcurrency(10) // 同时向 mxgate 写入数据的线程数量。
                .withRequestTimeoutMillis(3000) // 每个线程每次数据写入请求的超时时间(毫秒)。
                .withMaxRetryAttempts(3) // 每个线程每次写入请求遇到问题后的重试次数。
                .withRetryWaitDurationMillis(3000) // 每次重试的时间间隔（当前的实现，每次重试的时间间隔是固定的）。
                .withRequestType(RequestType.WithHTTP) // SDK 支持通过 HTTP 和 gRPC 两种方式向 mxgate post 数据，对应的配置为：RequestType.WithHTTP，RequestType.WithGRPC
                .withCircuitBreaker() // 使用内置熔断器。若失败率或者慢请求率达到阈值，则会开启，开启后持续30秒，暂停向 mxgate 发送数据，亦无法 append tuple
                .withMinimumNumberOfCalls(1) // 熔断器生效的最小请求数(要求 >= 1，默认 10)
                .withSlidingWindowSize(10) // 用以计算失败率的滑动窗口大小（要求 >= 1，默认 100）
                .withFailureRateThreshold(60.0f) // 失败率阈值（要求 >0 且 <= 100），若失败率达到阈值则会开启熔断器
                .withSlowCallDurationThresholdMillis(1000) // 慢请求时长阈值（毫秒），超过该时长则认为是慢请求（注意该时长应小于请求超时时间）
                .withSlowCallRateThreshold(80.0f)// 慢请求阈值，若慢请求率达到阈值则会开启熔断器
                //  .withRequestAsync(true)// 开启异步模式向 mxgate 发送数据的工作者线程（如使用此功能去掉此行注释内容即可）
                //  .withConcurrency(20)// 通常异步模式只需要几十的并发，就可以达到和同步模式同样甚至更高的吞吐（如使用此功能去掉此行注释内容即可）
                .build();

        //  builder.getTupleCacheSize(); // 用于实时获取 SDK 内部 Tuple cache 中剩余 Tuple 的数量（如使用此功能去掉此行注释内容即可）


        builder.connect("http://172.16.100.68:8086/", "172.16.100.68:8087", "public", "mxgate", new ConnectionListener() {
            public void onSuccess(MxClient client) {

                // 循环调用for即可
                // client.withIntervalToFlushMillis(1000);
                //                client.withEnoughBytesToFlush(3000000);

                Tuple tuple1 = client.generateEmptyTuple();
                Tuple tuple2 = client.generateEmptyTuple();

                tuple1.addColumn("ts", "2022-05-18 16:30:06");
                tuple1.addColumn("tag", 102020030);
                tuple1.addColumn("c1", 1.1);
                tuple1.addColumn("c2", 2.2);
                tuple1.addColumn("c3", 3.3);
                tuple1.addColumn("c4", 4.4);
                tuple1.addColumn("c5", "中文字符测试-1");
                tuple1.addColumn("c6", "lTxFCVLwcDTKbNbjau_c6");
                tuple1.addColumn("c7", "lTxFCVLwcDTKbNbjau_c7");
                tuple1.addColumn("c8", "lTxFCVLwcDTKbNbjau_c8");

                tuple2.addColumn("ts", "2022-05-18 16:30:06");
                tuple2.addColumn("tag", 102020030);
                tuple2.addColumn("c1", 1.1);
                tuple2.addColumn("c2", 2.2);
                tuple2.addColumn("c3", 3.3);
                tuple2.addColumn("c4", 4.4);
                tuple2.addColumn("c5", "中文字符测试-2");
                tuple2.addColumn("c6", "lTxFCVLwcDTKbNbjau_c26");
                tuple2.addColumn("c7", "lTxFCVLwcDTKbNbjau_c27");
                tuple2.addColumn("c8", "lTxFCVLwcDTKbNbjau_c28");

                client.appendTuples(tuple1, tuple2);

                // sendData(client);
            }

            public void onFailure(String failureMsg) {

            }
        });

    }

    private void sendData(MxClient client) {
        if (client != null) {
            // MxClient 会累积一批 Tuple 作为一个微批一并发送给 mxgate，
            // 该 API 会设置每个微批累积的等待时间，默认 2000 millisecond，
            // 也就是每个 2s 会尝试发送一批数据到 mxgate，如果该 2s 的时间段内
            // 没有数据写入，则不会发送。
            client.withIntervalToFlushMillis(5000);
            // 设置累积多少个 Tuple 字节作为一个微批进行发送，即使时间没有达到
            // 设置的 flush interval，累积的微批字节数达到了，也会发送。
            // flush interval 的时间达到了，没有累积够设定的字节数，也会发送。
            client.withEnoughBytesToFlush(500);
            // 相比 withEnoughBytesToFlush，在 appendTuple 的时候
            // 性能得到提升，因为这样避免了计算字节数
            // 视具体使用的场景而定，如果 withEnoughBytesToFlush
            // 同样能满足性能要求，那么每次 flush，数据量将会更加均匀。
            // withEnoughLinesToFlush 的优先级会高于 withEnoughBytesToFlush
            client.withEnoughLinesToFlush(2000);
            // 每个 MxClient 独占一个私有的对象池
            // 使用的时候，需要根据每次 MxClient flush 的 Tuple 数量合理的设置对象池的大小
            // client.useTuplesPool(poolSize); //（如使用此功能去掉此行注释内容即可 ）
            // MxClient 支持压缩，需配合 mxgate v4.7.6 及更高版本一起使用
            // client.withCompress(); //（如使用此功能去掉此行注释内容即可）
            // 对于 HTTP 的请求，可以不使用 base64 encoding，gRPC 则需要进行 base64 encoding。
            // client.withBase64Encode4Compress();  // （如使用此功能去掉此行注释内容即可）
            // MxClient 可以注册一个 DataPostListener，每批数据发送的成功和失败
            // 都会在 onSuccess 和 onFailure 中回调。你可以了解哪些 Tuple 写入成功，哪些 Tuple 写入失败
            client.registerDataPostListener(new DataPostListener() {

                public void onSuccess(Result result) {
                    System.out.println("Send tuples success: " + result.getMsg());
                    System.out.println("Succeed lines onSuccess callback " + result.getSucceedLines());
                }

                public void onFailure(Result result) {
                    // result.getErrorTuplesMap() 包含错误的行和错误原因的键值对 Tuple -> String
                    // Key 是错误的行， value 是错误的原因；
                    for (Map.Entry<Tuple, String> entry : result.getErrorTuplesMap().entrySet()) {
                        // log.error("error tuple of table={}, tuple={}, reason={}", entry.getKey().getTableName(), entry.getKey(), entry.getValue());
                        for (Column c : entry.getKey().getColumns()) {
                            //   l.error("error entry columns {} = {}", c.getColumnName(), c.getValue());
                        }
                    }
                    // System.out.println(result.getSuccessLines());
                }
            });


        }
    }
}
